package org.sn.jdish.spark.operator;

import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.VoidFunction;

/**
 * 合并两个RDD，不去重， 要求两个RDD中的元素类型一致------逻辑上抽象的合并
 * 
 * @author snzigod@hotmail.com
 *
 */
public class Union {

	public static void main(String[] args) {
		/**
		 * SparkConf:第一步创建一个SparkConf，在这个对象里面可以设置允许模式Local Standalone yarn
		 * AppName(可以在Web UI中看到) 还可以设置Spark运行时的资源要求
		 */
		SparkConf conf = new SparkConf().setAppName("sparkOperator").setMaster("local");

		/**
		 * 基于SparkConf的对象可以创建出来一个SparkContext Spark上下文
		 * SparkContext是通往集群的唯一通道，SparkContext在创建的时候还会创建任务调度器
		 */
		JavaSparkContext sc = new JavaSparkContext(conf);

		union(sc);

		sc.close();
	}

	public static void union(JavaSparkContext sc) {
		List<String> datas1 = Arrays.asList("张三", "李四", "tom");
		List<String> datas2 = Arrays.asList("张三", "tom", "gim");

		JavaRDD<String> data1RDD = sc.parallelize(datas1);
		JavaRDD<String> data2RDD = sc.parallelize(datas2);

		/**
		 * 合并两个RDD，不去重，要求两个RDD中的元素类型一致。
		 */
		JavaRDD<String> dataUnionAll = data1RDD.union(data2RDD);

		dataUnionAll.foreach(new VoidFunction<String>() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(String pair) throws Exception {
				System.out.println(pair);
			}
		});

		JavaRDD<String> dataUnion = dataUnionAll.distinct();

		dataUnion.foreach(new VoidFunction<String>() {
			private static final long serialVersionUID = 1L;

			@Override
			public void call(String pair) throws Exception {
				System.out.println(pair);
			}
		});
	}

}
